package net.hn.hnms.biz.upkafka.service;

import net.hn.hnms.biz.upkafka.config.KafkaProperties;
import net.hn.hnms.biz.upkafka.config.SpringContextUtil;
import net.hn.hnms.biz.upkafka.domain.SynergiaNowPo;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class AQJKProducer extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);

    public final KafkaProducer<String, String> producer;

    private final String topic;
    private final String mineCode;

    public final Boolean isAsync;
    public AQJKProducer(String topicName, Boolean asyncEnable,String mineCode) {
        Properties props = Producer.initProperties(topicName,mineCode);
        producer = new KafkaProducer<String, String>(props);
        topic = topicName;
        isAsync = asyncEnable;
        this.mineCode = mineCode;
    }
    @Override
    public void run() {
        FTPClientProducer ftpClientProducer = new FTPClientProducer();
        String ftpServer = "";  // 替换为实际的 FTP 服务器地址
        String aqjkUsername = "";  // 替换为实际的用户名
        String aqjkPassword = "";  // 替换为实际的密码
        String ftpDirectory = "";  // 指定要访问的 FTP 文件夹目录

        String rydwUsername = "";  // 替换为实际的用户名
        String rydwPassword = "";
        if ("150621011015".equals(mineCode)) {
            //高头窑煤矿
            ftpServer = "10.33.171.130";

            aqjkUsername = "hnftp_aqjk";
            aqjkPassword = "hnftp@2024";

            ftpDirectory = "";

            rydwUsername = "hnftp";
            rydwPassword = "hnftp@2024";
        }else if ("150781013671".equals(mineCode)) {
            //灵露煤矿
            ftpServer = "10.41.105.108";

            aqjkUsername = "aqjk";
            aqjkPassword = "linglu@2024";

            ftpDirectory = "";

            rydwUsername = "rydw";  // 替换为实际的用户名
            rydwPassword = "linglu@2024";
        }
        // 构造消息记录
        while (true) {
            //获取安全监控数据
            //CDDY  测点定义
            List<String> cddyList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "CDDY");
            this.sendMessage(KafkaProperties.AQJC_CDDY_TOPIC,mineCode, cddyList,"CDDY");
            //CDSS  测点实时
            List<String> cdssList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "CDSS");
            this.sendMessage(KafkaProperties.AQJC_CDSS_TOPIC,mineCode, cdssList,"CDSS");
            //FZDY 分站定义
            List<String> fzdyList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "FZDY");
            this.sendMessage(KafkaProperties.AQJC_FZDY_TOPIC,mineCode, fzdyList,"FZDY");
            //FZSS 分站实时
            List<String> fzssList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "FZSS");
            this.sendMessage(KafkaProperties.AQJC_FZSS_TOPIC,mineCode, fzssList,"FZSS");
            //TJSJ  统计数据
            List<String> tjsjList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "TJSJ");
            this.sendMessage(KafkaProperties.AQJC_TJSJ_TOPIC,mineCode, tjsjList,"TJSJ");
            //YCBJ 异常报警
            List<String> ycbjList = ftpClientProducer.readFtpFiles(ftpServer, aqjkUsername, aqjkPassword, ftpDirectory, "YCBJ");
            this.sendMessage(KafkaProperties.AQJC_YCBJ_TOPIC,mineCode, ycbjList,"YCBJ");


             
            //获取人员定位数据
            //JZSS  基站实时
            List<String> jzssList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "JZSS");
            this.sendMessage(KafkaProperties.RYDW_JZSS_TOPIC,mineCode, jzssList,"JZSS");
            //RYJZ 人员基站
            List<String> ryjzList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYJZ");
            this.sendMessage(KafkaProperties.RYDW_RYJZ_TOPIC,mineCode, ryjzList,"RYJZ");
            //RYQY 人员区域
            List<String> ryqyList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYQY");
            this.sendMessage(KafkaProperties.RYDW_RYQY_TOPIC,mineCode, ryqyList,"RYQY");
            //RYSS 人员实时
            List<String> ryssList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYSS");
            this.sendMessage(KafkaProperties.RYDW_RYSS_TOPIC,mineCode, ryssList,"RYSS");
            //RYXX 人员信息
            List<String> ryxxList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYXX");
            this.sendMessage(KafkaProperties.RYDW_RYXX_TOPIC,mineCode, ryxxList,"RYXX");
            //RYCS 人员超时
            List<String> rycsList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYCS");
            this.sendMessage(KafkaProperties.RYDW_RYCS_TOPIC,mineCode, rycsList,"RYCS");
            //RYCY 人员超员
            List<String> rycyList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYCY");
            this.sendMessage(KafkaProperties.RYDW_RYCY_TOPIC,mineCode, rycyList,"RYCY");
            //RYQJ 人员求救
            List<String> ryqjList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYQJ");
            this.sendMessage(KafkaProperties.RYDW_RYQJ_TOPIC,mineCode, ryqjList,"RYQJ");
            //RYXZ 进入限制区域
            List<String> ryxzList = ftpClientProducer.readFtpFiles(ftpServer, rydwUsername, rydwPassword, ftpDirectory, "RYXZ");
            this.sendMessage(KafkaProperties.RYDW_RYXZ_TOPIC,mineCode, ryxzList,"RYXZ");
            try {
                // 线程阻塞 1 分钟 (60000 毫秒)
                Thread.sleep(60000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }

    public void sendMessage(String topicString,String mineCode, List<String> list,String type) {
        String key = String.valueOf(mineCode);
        if (list.isEmpty()) {
            System.out.println(type+"  type : " + "No Data");
            return;
        }
        //发送获取到的数据
        list.forEach(str->{
            //获取头部信息
            String[] split = str.split("~");
            String topMessage = "";
            if (split.length > 1) {
                topMessage = split[0];
            } else {
                topMessage = str;
            }
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(topicString, key, str.toString());
            if (isAsync) {
                // 异步发送
                producer.send(record);
            } else {
                try {
                    // 同步发送
                    producer.send(record).get();
                    System.out.println(type+"发送成功:" + topMessage);
                } catch (InterruptedException ie) {
                    LOG.info("The InterruptedException occured : {}.", ie);
                } catch (ExecutionException ee) {
                    LOG.info("The ExecutionException occured : {}.", ee);
                }
            }
        });
    }
}